package com.feeyo.groundservice.base.influx;

import java.util.concurrent.TimeUnit;

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Pong;

import com.feeyo.groundservice.base.util.PropertiesUtils;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.json.JsonArray;

public class Influx extends AbstractVerticle {
	public static InfluxDB client;
	public static BatchPoints batchPoints;
	private static String dbName = PropertiesUtils.getProperty("influx.database");
	private static String measure = PropertiesUtils.getProperty("influx.table");
	private static final long interval = 1000 * 60  * 60 *24 ;
	@Override
	public void start(Future<Void> startFuture) throws Exception {
//		client = InfluxDBFactory.connect("http://"+PropertiesUtils.getProperty("influx.host")+
//				":"+PropertiesUtils.getProperty("influx.port"), PropertiesUtils.getProperty("influx.user"), PropertiesUtils.getProperty("influx.password"));
		client = InfluxDBFactory.connect("http://"+PropertiesUtils.getProperty("influx.host")+
				":"+PropertiesUtils.getProperty("influx.port"));
		client.createDatabase(dbName);
		//client.enableBatch(2, 100, TimeUnit.MILLISECONDS);
		client.enableBatch(20, 200, TimeUnit.MILLISECONDS);
		boolean influxDBstarted = false;
		do {
			Pong response;
			try {
				response = client.ping();
				if (!response.getVersion().equalsIgnoreCase("unknown")) {
					influxDBstarted = true;
				}
			} catch (Exception e) {
				// NOOP intentional
				e.printStackTrace();
			}
			Thread.sleep(100L);
		} while (!influxDBstarted);
		
		System.out.println("################################################################################## ");
		System.out.println("#  Connected to InfluxDB Version: " + client.version() + " #");
		System.out.println("#  describeDatabases: " + client.describeDatabases() + " #");
		System.out.println("##################################################################################");
		
		
//		batchPoints = BatchPoints  
//	            .database(dbName)  
//	            //.tag("async", "true") //Add a tag to this set of points.  
//	            .retentionPolicy("autogen")  
//	            .consistency(InfluxDB.ConsistencyLevel.ONE)  
//	            .build();  
		
        vertx.eventBus().consumer("tots", msg->{
            JsonArray ja = (JsonArray)msg.body();
            String devid = ja.getString(0);
            //System.out.println("devid:"+devid);
            
            long devTime = ja.getLong(7);
            long serverTime = System.currentTimeMillis();
            if ((serverTime - devTime) > interval)
            	;
            else{
	    	    Point point = Point.measurement(measure)  
	    	    		.time(ja.getLong(7), TimeUnit.MILLISECONDS)
	    	            .tag("devid", devid)
	    	            .addField("lat", ja.getDouble(1))  
	    	            .addField("lon", ja.getDouble(2))  
	    	            .addField("speed", ja.getDouble(3))
	    	            .addField("direct", ja.getDouble(4))
	    	            .addField("location", true)
	    	            .addField("ntype", ja.getString(8))
	    	            .build();
	    	    
	    	    //batchPoints.point(point);  
	    	    client.write(dbName,"autogen",point);
            }
    	    //System.out.println("batchPoint.size():"+batchPoints.getPoints().size());
    	    //ja = null;
    	    //point = null;
//    	    client.write(dbName, "autogen", point);
//    	    Query query = new Query("SELECT * FROM pos_2", dbName);  
//            QueryResult result = client.query(query); 
//            System.out.println("result:"+result.getResults().get(0).toString());
        });
		
        startFuture.complete();
	}

	/**
	 * devid
	 * lat,lon,dir,alt,HORACC,VERACC,
	 * <p>Description: <／p>
	 */
	public static void writeData001() {  
	    Point point001 = Point.measurement("pos")  
	            .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)  
	            .tag("devid", "12333")
	            .addField("lat", 12.11)  
	            .addField("lon", 22.11)  
	            .addField("speed", 32.3)
	            .addField("direct", 180.1)
	            .addField("location", true)
	            .build();  
	    
	  
	    batchPoints.point(point001);  
	    client.write(batchPoints);  
	}  
}
